package widefinder

import java.nio.ByteBuffer
import java.nio.channels.FileChannel
import java.util.concurrent.*
import java.util.regex.Matcher
import java.util.regex.Pattern
import groovy.util.concurrent.AtomicMap
import java.util.concurrent.atomic.AtomicLong
import groovy.util.concurrent.AtomicMapEntry

@Typed
class Start2
{
   /**
    * Top N counter
    */
    static final int N = 10;

    private static final String  ARTICLE_PREFIX  = '/ongoing/When/';
    private static final Pattern pattern = Pattern.compile("^$ARTICLE_PREFIX\\d{3}x/\\d{4}/\\d{2}/\\d{2}/[^ .]+\$")

    /**
     * Character constants
     */
    private static final byte CR    = 0x0D; // "\r"
    private static final byte LF    = 0x0A; // "\n"
    private static final byte SPACE = 0x20; // " "


    public static void main ( String ... args )
    {
        final File file    = new File( args[ 0 ] );
        assert file.isFile(), "File [$file] is not available" ;

        final long               t          = System.currentTimeMillis();
        final ThreadPoolExecutor pool       =
            ( ThreadPoolExecutor ) Executors.newFixedThreadPool(200);

        processFile( file, pool );

        pool.shutdown();
        pool.awaitTermination(30,TimeUnit.SECONDS)

        println "[${ System.currentTimeMillis() - t }] ms"
    }



   /**
    * Reports all results
    */
    private static void reportTopResults ( int n, ThreadPoolExecutor pool )
    {
        List<Future> futures = [];
        pool.getCorePoolSize().times
        {
            /**
             * Each thread calculates it's own "top n" maps
             */
//            futures << pool.submit((Callable){ (( Stat ) Thread.currentThread()).calculateTop( n ) })
        }

//        List<List<Map<Long, Collection<String>>>> topMaps           = futures*.get()
//        Map<String, Long>                         topArticlesToHits = StatUtils.sumAndTop( n, topMaps*.get( 0 ));
//        Map<String, Long>                         topUrisToBytes    = StatUtils.sumAndTop( n, topMaps*.get( 1 ));
//        Map<String, Long>                         topUrisTo404      = StatUtils.sumAndTop( n, topMaps*.get( 2 ));

        futures = [];
        pool.getCorePoolSize().times
        {
            /**
             * Each thread calculates it's own "top n clients/referrers" maps
             * (according to "top articles" calculated previously)
             */
//            futures << pool.submit((Callable){ (( Stat ) Thread.currentThread()).filterWithArticles( topArticlesToHits.keySet()) })
        }

//        List<List<Map<String, L>>> topArticlesMaps        = futures*.get();
//        Map<String, Long>          topClientsToArticles   = StatUtils.sumAndTop2( n, topArticlesMaps*.get( 0 ));
//        Map<String, Long>          topReferrersToArticles = StatUtils.sumAndTop2( n, topArticlesMaps*.get( 1 ));
//
//        report( "Top $n articles (by hits)",          topArticlesToHits      );
//        report( "Top $n URIs (by bytes count)",       topUrisToBytes         );
//        report( "Top $n URIs (by 404 responses)",     topUrisTo404           );
//        report( "Top $n clients (by top articles)",   topClientsToArticles   );
//        report( "Top $n referrers (by top articles)", topReferrersToArticles );
    }


    static void report( String title, Map<String, Long> map )
    {
        println ">>> $title <<<: \n* ${ map.entrySet().collect{ Map.Entry entry -> "${ entry.key } : ${ entry.value }" }.join( "\n* " ) }"
    }

    static final ByteBuffer stop = ByteBuffer.allocate(1)

    static final Stat globalStat = []

   /**
    * Reads number of lines in the channel specified
    */
    private static void processFile ( File file, ThreadPoolExecutor pool )
    {
        final def fis     = new FileInputStream( file );
        final def channel = fis.getChannel();

        int n = Runtime.runtime.availableProcessors()-1

        def queue       = new ArrayBlockingQueue<ByteBuffer>(2*n)
        def freeBuffers = new ArrayBlockingQueue<ByteBuffer>(2*n)

        for (i in 0..<n) {
            pool.execute {
//                while (true) {
//                    def buffer = queue.take()
//                    if (buffer === stop)
//                        break;

//                    def s = System.currentTimeMillis()
//                    processLines(buffer.array(), 0, buffer.position(), globalStat)
//                    println "process: " + (System.currentTimeMillis() - s)

//                    freeBuffers << buffer
//                }

//                List<Map<Long, Collection<String>>> topMap = ((Stat)Thread.currentThread()).calculateTop( n );
//
//                Map<String, Long>                         topArticlesToHits = StatUtils.sumAndTop( n, [topMap.get( 0 )])
//                Map<String, Long>                         topUrisToBytes    = StatUtils.sumAndTop( n, [topMap.get( 1 )])
//                Map<String, Long>                         topUrisTo404      = StatUtils.sumAndTop( n, [topMap.get( 2 )])
//
//                List<Map<String, L>> topArticlesMaps = (( Stat ) Thread.currentThread()).filterWithArticles( topArticlesToHits.keySet())
//
//                Map<String, Long>          topClientsToArticles   = StatUtils.sumAndTop2( n, [topArticlesMaps.get( 0 )])
//                Map<String, Long>          topReferrersToArticles = StatUtils.sumAndTop2( n, [topArticlesMaps.get( 1 )])
//
//                report( "Top $n articles (by hits)",          topArticlesToHits      );
//                report( "Top $n URIs (by bytes count)",       topUrisToBytes         );
//                report( "Top $n URIs (by 404 responses)",     topUrisTo404           );
//                report( "Top $n clients (by top articles)",   topClientsToArticles   );
//                report( "Top $n referrers (by top articles)", topReferrersToArticles );
            }
         }

        for(i in 0..<(2*n))
            freeBuffers << ByteBuffer.allocate( 512*1024 );

        long size = channel.size()

        for (i in 0..<30)
        pool.execute {
            Matcher ARTICLE_MATCHER = pattern.matcher( "" );
            ByteBuffer prevBuffer = null
            while (true) {
                ByteBuffer buffer = queue.take()
//                def s = System.currentTimeMillis()
                if (buffer === stop)
                    break;

                int start = findFirstEol(buffer)
                int end   =  findLastEol(buffer)
                processLines(buffer.array(), start, end, globalStat, ARTICLE_MATCHER)

//                println ("process: " + (System.currentTimeMillis() - s))
                freeBuffers << buffer
            }
        }


        while (channel.position() < size)
        {
            final ByteBuffer buffer = freeBuffers.take ()
//            def s = System.currentTimeMillis()
            buffer.rewind()
            channel.read(buffer)
//            println ("read: " + (System.currentTimeMillis() - s))
            queue << buffer
        }

        println "reading finished..."

        for (i in 0..<n)
            queue << stop

        channel.close();
        fis.close();
    }

    static int findLastEol(ByteBuffer buffer) {
        def array = buffer.array()
        def end = buffer.position() - 1
        for ( ; end >= 0; end--) {
            if (array[end] == LF || array[end] == CR) {
                break;
            }
        }
        end < 0 ? buffer.position() : end
    }

    static int findFirstEol(ByteBuffer buffer) {
        def array = buffer.array()
        def start = 0
        for ( ; start != buffer.position(); start++) {
            if (array[start] == LF || array[start] == CR) {
                break;
            }
        }

        start == buffer.position () ? 0 : start
    }

    /**
    * This is where each thread gets it's own byte[] chunk to analyze:
    * - it starts at index "startIndex"
    * - it ends   at index "endIndex" - 1
    * - it contains a number of complete rows (no half rows)
    */
    static void processLines( byte[] array, int startIndex, int endIndex, Stat stat, Matcher matcher )
    {
        String  clientAddress = null;
        String  httpMethod    = null;
        String  uri           = null;
        String  statusCode    = null;
        String  byteCount     = null;
        String  referrer      = null;

        int     tokenStart    = startIndex;
        int     tokenCounter  = 0;

        for ( int index = startIndex; index < endIndex; index++ ) // "index" is incremented manually - Range doesn't fit here
        {
            if ( space( array[ index ] ))
            {
                switch ( tokenCounter++ )
                {
                    case 0  : clientAddress = string( array, tokenStart, index );
                              break;
                    case 5  : httpMethod    = string( array, tokenStart + 1, index ); // Getting rid of starting '"'
                              break;
                    case 6  : uri           = string( array, tokenStart, index );
                              break;
                    case 8  : statusCode    = string( array, tokenStart, index );
                              break;
                    case 9  : byteCount     = string( array, tokenStart, index );
                              break;
                    case 10 : referrer      = string( array, tokenStart + 1, index - 1 ); // Getting rid of wrapping '"'
                              break;
                }

                if ( referrer == null )
                {
                    /**
                     * Not all tokens are found yet - keep looking for the next one
                     * (skipping "space" sequence)
                     */
                    while ( space( array[ index ] )){ index++ }
                    tokenStart = index;
                }
                else
                {
                    /**
                     * We've found all tokens ("referrer" was the last one) - updating statistics,
                     * adding the data read from the current line
                     */
                    stat.update( clientAddress, httpMethod, uri, statusCode, byteCount, referrer, matcher );

                    /**
                     * Finding and skipping "end of line" sequence
                     */
                     while (( index < endIndex ) && ( ! endOfLine( array[ index ] ))){ index++ }
                     while (( index < endIndex ) && (   endOfLine( array[ index ] ))){ index++ }

                     tokenStart    = index;
                     tokenCounter  = 0;

                     clientAddress = null;
                     httpMethod    = null;
                     uri           = null;
                     statusCode    = null;
                     byteCount     = null;
                     referrer      = null;
                }
            }
        }
    }

    static class PseudoString {
        byte [] array
        int start, end

        int hashCode () {
            def h = end-start
            for (i in start..<end)
                h += 31*h + array[i]
        }

        boolean equals(Object o) {
            (o instanceof PseudoString)
        }
    }

   /**
    * Creates a String using the array specified and "UTF-8" charset.
    * String starts at index "start" and ends at index "end - 1"
    */
    private static String string( byte[] array, int start, int end )
    {
        new String( array, 0, start, ( end - start ) );
    }


   /**
    * Determines if byte specified is a space character
    */
    private static boolean space( byte b )
    {
        ( b == SPACE );
    }


   /**
    * Determines if byte specified is an end-of-line character
    */
    private static boolean endOfLine( byte b )
    {
        ( b == LF )
    }

    static class Stat
    {

       /**
        * Gets a counter bound to the key specified in the map passed
        * (creates a new one if it doesn't exist yet)
        */
        private static L get( Map<String, L> map, String key )
        {
            L      counter = map[ key ];
            if ( counter == null) { map[ key ] = (counter = new L()) }
            return counter;
        }


        /**
         * Gets a counter bound to the key1 => key2 specified in the map passed
         * (creates a new one if it doesn't exist yet)
         */
        private static L get( Map<String, Map<String, L>> map, String key, String secondKey )
        {
            Map<String, L> secondMap = map[ key ];
            if ( map[key] == null ) { map[ key ] = (secondMap = new HashMap<String, L>()) }
            return get( secondMap, secondKey );
        }

       /**
        * Maps holding all statistical data
        */
        ArticleMap                  articlesToHits      = []
        StringCounterMap            uriToByteCounts     = []
        StringCounterMap            uriTo404            = []


//       /**
//        * Statistical maps counters convenience accessors
//        */
//        private L articlesToHitsCounter      ( String articleUri )                       { get( this.@articlesToHits,      articleUri  ) }
//        private L uriToByteCountsCounter     ( String uri        )                       { get( this.@uriToByteCounts,     uri         ) }
//        private L uriTo404Counter            ( String uri        )                       { get( this.@uriTo404,            uri         ) }
//        private L clientsToArticlesCounter   ( String articleUri, String clientAddress ) { get( this.@articlesToClients,   articleUri, clientAddress ) }
//        private L referrersToArticlesCounter ( String articleUri, String referrer      ) { get( this.@articlesToReferrers, articleUri, referrer      ) }



       /**
        * Updates statistics according to benchmark needs:
        * - http://wikis.sun.com/display/WideFinder/The+Benchmark
        * - http://groovy.codehaus.org/Regular+Expressions
        */
        void update( String clientAddress, String httpMethod, String uri, String statusCode, String byteCount, String referrer, Matcher matcher )
        {
            boolean isArticle = (( httpMethod.equals('GET') )             &&
                                 ( uri.startsWith( ARTICLE_PREFIX )) &&
                                 ( ! uri.endsWith( '.png' ))         &&
                                 ( matcher.reset( uri ).lookingAt()));

            if ( isArticle )
            {
                addArticle( uri,
                            clientAddress,
                            ((( ! referrer.isEmpty()) && ( referrer.equals('-') )) ? referrer : null ));
            }

            int parseInt
            try { parseInt = Integer.parseInt(byteCount, 10) } catch (NumberFormatException e) {}
            addUri( uri,
                    ( byteCount.contains( '-' ) ? 0 : parseInt),
                    ( statusCode.equals('404') ));
        }


        /**
         * Adds new articles to statistics
         *
         * @param articleUri    URI of article
         * @param clientAddress address of requesting client
         * @param referrer      request referrer
         */
        private void addArticle( String articleUri, String clientAddress, String referrer )
        {
            ArticleMap.Entry article = articlesToHits[articleUri]
            article.incrementAndGet();
            synchronized(article.clients) {
                L cl = article.clients.get(clientAddress)
                if (cl == null) {
                    article.clients.put(clientAddress, cl = new L ())
                }
                cl.increment()
            }

            if ( referrer )
            {
                synchronized(article.referreres) {
                    L cl = article.referreres.get(clientAddress)
                    if (cl == null) {
                        article.referreres.put(clientAddress, cl = new L ())
                    }
                    cl.increment()
                }
            }
        }


        /**
         * Adds new URI to statistics
         *
         * @param uri   request URI
         * @param bytes request total bytes returned
         * @param is404 whether a response was 404
         */
        private void addUri( String uri, int bytes, boolean is404 )
        {
            if ( bytes > 0 ) { uriToByteCounts[uri].addAndGet( bytes ) }
            if ( is404     ) { uriTo404[uri].incrementAndGet()         }
        }



       /**
        * Calculates "top N" counters for:
        * - 'articles to hits' map
        * - 'uri to bytes' map
        * - 'uri to missed (404)' map
        *
        * Returns a list of 3 Map<Long, Collection<String>>.
        * Each map:
        * Key   - a counter in the "top N" counters
        * Value - collection of String values corresponding to that counter
        */
        List<Map<Long, Collection<String>>> calculateTop( int n )
        {
            List<Map<Long, Collection<String>>> result =
            [
//                StatUtils.topCountersMap( n, getArticlesToHits()),
//                StatUtils.topCountersMap( n, getUriToByteCounts()),
//                StatUtils.topCountersMap( n, getUriTo404())
            ]

            /**
             * Nuking raw data the moment it's no longer needed
             */
            setArticlesToHits ( null );
            setUriToByteCounts ( null );
            setUriTo404 ( null );

            return result;
        }



       /**
        * Filters article-related Maps:
        * - 'articles to clients' map
        * - 'articles to referrers' map
        *
        * Returns a list of 2 Map<String, L>
        * Each map:
        * Key   - client address / referrer (accessing one of "top articles")
        * Value - counter, a number of time this access was made
        */
        List<Map<String, L>> filterWithArticles( Set<String> topArticles )
        {
            List<Map<String, L>> result =
            [
//                filterWithArticles( topArticles, getArticlesToClients()),
//                filterWithArticles( topArticles, getArticlesToReferrers()),
            ]

            /**
             * Nuking raw data the moment it's no longer needed
             */
//            setArticlesToClients( null );
//            setArticlesToReferrers( null );

            return result;
        }



       /**
        * Filters an article-related "valuesMaps" (like map of clients or referrers) with "top articles" specified
        */
        private Map<String, L> filterWithArticles( Set<String> topArticles, Map<String, Map<String, L>> valuesMaps )
        {
            /**
             * Maps (String value => articles hits) of "top articles" accesses
             * One map for each "top article" - they're summarized below into one bigger Map<String, L>
             */
            List<Map<String, L>> topArticlesMaps = valuesMaps.keySet().
                                                  /**
                                                   * Filtering out URIs which are not from "top articles" Set
                                                   */
                                                   findAll{ String articleUri -> topArticles.contains( articleUri )}.
                                                  /**
                                                   * Collecting a Map<String, L> (String value => article hits counter)
                                                   * for each URI that is left after "top articles" filtering
                                                   */
                                                   collect{ String articleUri -> valuesMaps[ articleUri ] };

            /**
             * Result map (String value => articles hits):
             * summarizing all above (smaller) maps for all "top articles" (one smaller map per "top article")
             * into one bigger Map<String, L>
             */
            Map<String, L> resultMap = new HashMap<String,L>();
            topArticlesMaps.each
            {
                Map<String, L> map ->
                map.each
                {
                    String value, L counter ->

                    if ( ! resultMap[ value ] ) { resultMap[ value ] = new L() }
                    resultMap[ value ].add( counter.counter());
                }
            }

            return resultMap;
        }
    }
}